package transform;

import beans.SenSorReading;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Collections;

public class SplitAndSelect {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //从文件中读取数据
        DataStreamSource<String> inputStream = env.readTextFile("src/main/resources/sensor.txt");

        SingleOutputStreamOperator<SenSorReading> mapStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SenSorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        //分流，将数据按照三十度来进行分流
        SplitStream<SenSorReading> splitStream = mapStream.split(new OutputSelector<SenSorReading>() {
            @Override
            public Iterable<String> select(SenSorReading senSorReading) {
                return (senSorReading.getTemperature() > 30 ? Collections.singletonList("high") : Collections.singletonList("low"));
            }
        });
        DataStream<SenSorReading> highStream = splitStream.select("high");
        DataStream<SenSorReading> lowStream = splitStream.select("low");
        DataStream<SenSorReading> allStream = splitStream.select("high", "low");

        highStream.print("high");
        lowStream.print("low");
        allStream.print("all");

        env.execute();
    }
}
